package com.xyzq.simpson.hibbert.storage.influx;

import com.xyzq.simpson.base.text.Text;
import com.xyzq.simpson.base.type.Table;
import com.xyzq.simpson.base.type.core.ILink;
import okhttp3.OkHttpClient;
import org.influxdb.BatchOptions;
import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.*;
import org.influxdb.impl.InfluxDBMapper;
import org.influxdb.impl.InfluxDBResultMapper;
import org.influxdb.querybuilder.SelectQueryImpl;
import org.influxdb.querybuilder.SelectionQueryImpl;
import org.influxdb.querybuilder.WhereQueryImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static org.influxdb.querybuilder.BuiltQuery.QueryBuilder.*;
import static org.influxdb.querybuilder.time.DurationLiteral.MILLISECONDS;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @Description:
 * @date 2020/3/3 16:53
 */
public class InfluxDBConnection<T> {

    private static final Logger logger = LoggerFactory.getLogger(InfluxDBConnection.class);
    // 用户名
    private String username;
    // 密码
    private String password;
    // 连接地址
    private String openurl;
    // 数据库
    private String database;
    // 保留策略
    private String retentionPolicy;

    private InfluxDB influxDB;

    private final BatchOptions batchOptions = BatchOptions.DEFAULTS;

    public InfluxDBConnection(String username, String password, String openurl, String database,
                              String retentionPolicy) {
        this.username = username;
        this.password = password;
        this.openurl = openurl;
        this.database = database;
        this.retentionPolicy = retentionPolicy == null || retentionPolicy.equals("") ? "autogen" : retentionPolicy;
        this.batchOptions.actions(1000).bufferLimit(2000).flushDuration(100).jitterDuration(10).exceptionHandler((points, e) -> {
            List<Point> target = new ArrayList<>();
            points.forEach(target::add);
            String msg = String.format("failed to write points:%s\n", target.toString().substring(0, 10000));
            logger.error(msg, e);
        }).threadFactory(
                Executors.defaultThreadFactory()
        );
        influxDbBuild();
    }

    public InfluxDB influxDbBuild() {
        if (influxDB == null) {
            influxDB = InfluxDBFactory.connect(openurl, username, password,
                    new OkHttpClient.Builder()
                        .connectTimeout(1, TimeUnit.MINUTES)
                        .readTimeout(1, TimeUnit.MINUTES)
                        .writeTimeout(1, TimeUnit.MINUTES)
                        .retryOnConnectionFailure(true))
                    .enableBatch(batchOptions);
        }
        try {
            if (!this.databaseExist(database)) {
                createDB(database);
            }
        } catch (Exception e) {

        } finally {
            influxDB.setRetentionPolicy(retentionPolicy);
        }
        influxDB.setLogLevel(InfluxDB.LogLevel.NONE);
        return influxDB;
    }

    public boolean databaseExist(String database) {
        return influxDB.databaseExists(database);
    }

    /**
     * 创建数据库
     *
     * @param dbName
     */
    @SuppressWarnings("deprecation")
    public void createDB(String dbName) {
        influxDB.createDatabase(dbName);
    }

    /**
     * 删除数据库
     *
     * @param dbName
     */
    @SuppressWarnings("deprecation")
    public void deleteDB(String dbName) {
        influxDB.deleteDatabase(dbName);
    }

    /**
     * 测试连接是否正常
     *
     * @return true 正常
     */
    public boolean ping() {
        boolean isConnected = false;
        Pong pong;
        try {
            pong = influxDB.ping();
            if (pong != null) {
                isConnected = true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return isConnected;
    }

    /**
     * 创建自定义保留策略
     *
     * @param policyName
     *            策略名
     * @param duration
     *            保存天数
     * @param replication
     *            保存副本数量
     * @param isDefault
     *            是否设为默认保留策略
     */
    public void createRetentionPolicy(String policyName, String duration, int replication, Boolean isDefault) {
        String sql = String.format("CREATE RETENTION POLICY \"%s\" ON \"%s\" DURATION %s REPLICATION %s ", policyName,
                database, duration, replication);
        if (isDefault) {
            sql = sql.concat(" DEFAULT");
        }
        this.query(sql);
    }

    public void dropRetentionPolicy(String database, String retentionPolicy) {
        String sql = String.format("DROP RETENTION POLICY %s ON %s", retentionPolicy, database);
        this.query(sql);
    }

    public void createContinuousQuery(String measurement) {
        String cqName = String.format("cq_%s", measurement);
        String originMeasurement = String.format("%s.%s.%s", database, retentionPolicy, measurement);
        String cqMeasurement = String.format("%s.%s.%s_hour", database, retentionPolicy, measurement);
        String sql = String.format("CREATE CONTINUOUS QUERY \"%s\" ON %s RESAMPLE EVERY 1h FOR 2h BEGIN SELECT MEAN(*) INTO %s FROM %s GROUP BY time(1h),* FILL(none) END",
                cqName, database, cqMeasurement, originMeasurement);
        this.query(sql);
    }

    public boolean continuousQueryExists(String measurement) {
        String cqName = String.format("cq_%s", measurement);
        return continuousQueryExists(database, cqName);
    }

    public boolean continuousQueryExists(String database, String cqName) {
        String sql = "SHOW CONTINUOUS QUERIES";
        QueryResult result = query(sql);
        List<QueryResult.Series> seriesList = result.getResults().get(0).getSeries();
        if (seriesList != null) {
            for (QueryResult.Series series : seriesList) {
                if (database.equals(series.getName())) {
                    List<List<Object>> continuousQueryList = series.getValues();
                    if (continuousQueryList == null) {
                        return false;
                    } else {
                        for (List<Object> queryResult : continuousQueryList) {
                            if (cqName.equals(queryResult.get(0))) {
                                return true;
                            }
                        }
                    }
                }
            }
        }
        return false;
    }

    public void dropContinuousQuery(String databaseName, String cqName) {
        String sql = String.format("DROP CONTINUOUS QUERY %s ON %s", cqName, databaseName);
        query(sql);
    }

    public boolean measurementsExists(String measurement) {
        return measurementsExists(database, measurement);
    }

    public boolean measurementsExists(String database, String measurement) {
        String sql = String.format("SHOW MEASUREMENTS ON %s", database);
        QueryResult result = query(sql);
        if (result != null) {
            List<QueryResult.Series> seriesList = result.getResults().get(0).getSeries();
            if (seriesList != null) {
                QueryResult.Series series = seriesList.get(0);
                List<List<Object>> valueList = series.getValues();
                for (List<Object> value : valueList) {
                    if (measurement.equals(value.get(0))) {
                        return true;
                    }
                }
            }
        }
        return false;
    }

    /**
     * 查询
     *
     * @param command
     *            查询语句
     * @return
     */
    public QueryResult query(String command) {
        return influxDB.query(new Query(command, database));
    }

    public List<T> query(String command, Class<T> clazz) {
        QueryResult result = influxDB.query(new Query(command, database));
        InfluxDBResultMapper resultMapper = new InfluxDBResultMapper();
        return resultMapper.toPOJO(result, clazz);
    }

    public List<T> queryByMapper(String command, Class<T> clazz) {
        InfluxDBMapper influxDBMapper = new InfluxDBMapper(influxDB);
        Query query = new Query(command, database);
        return influxDBMapper.query(query, clazz);
    }

    public List<T> queryByMapper(Class<T> clazz) {
        InfluxDBMapper influxDBMapper = new InfluxDBMapper(influxDB);
        return influxDBMapper.query(clazz);
    }

    public void saveByMapper(T t) {
        InfluxDBMapper influxDBMapper = new InfluxDBMapper(influxDB);
        influxDBMapper.save(t);
    }

    /**
     * 查询
     *
     * @param command sql
     *
     * @param params 参数绑定
     *
     */
    public QueryResult queryByCommandAndParam(String command, Table<String, Object> params) {
        BoundParameterQuery.QueryBuilder queryBuilder = BoundParameterQuery.QueryBuilder.newQuery(command)
                .forDatabase(database);
        for(ILink<String, Object> link : params) {
            queryBuilder.bind(link.origin(), link.destination());
        }
        return influxDB.query(queryBuilder.create());
    }

    /**
     * 查询
     *
     * @param measurement
     *
     * @param param 不要带time参数
     *
     */
    public List<Table<String, Object>> queryByParam(String measurement, Table<String, Object> param, Set<String> tags) {
        return queryByParamWithLimit(measurement, param, -1, tags);
    }

    /**
     * 查询
     *
     * @param measurement
     *
     * @param param 不要带time参数
     *
     * @param limit
     */
    public List<Table<String, Object>> queryByParamWithLimit(String measurement, Table<String, Object> param, int limit, Set<String> tags) {
        WhereQueryImpl<SelectQueryImpl> where = select().from(database, measurement).where();
        buildQryWhereWithParam(where, param, tags);
        Query query = where;
        if(limit != 0) {
            query = where.limit(limit);
        }
        return convert(influxDB.query(query));
    }


    /**
     * 查询
     *
     * @param measurement
     *
     * @param param 不要带time参数
     *
     * @param startTi 开始时间
     *
     * @param endTi 结束时间
     */
    public List<Table<String, Object>> queryByParamBetweenTis(String measurement, Table<String, Object> param, long startTi, long endTi, Set<String> tags) {
        WhereQueryImpl<SelectQueryImpl> where = select().from(database, measurement).where();
        buildQryWhereWithParam(where, param, tags);
        where.and(gte("time",ti(startTi, MILLISECONDS)))
                .and(lte("time",ti(endTi, MILLISECONDS)));
        Query query = where;
        return convert(influxDB.query(query));
    }

    /**
     * 查询间隔时间聚合
     *
     * @param measurement
     *
     * @param param 不要带time参数
     *
     * @param startTi 开始时间
     *
     * @param endTi 结束时间
     *              *
     * @param tags
     *
     * @param timeInterval 间隔时长
     *
     * @param durationLiteral 间隔单位
     */
/*    public List<Table<String, Object>> queryMeanBetweenTisWithGroupByTi(String measurement, Table<String, Object> param,
                          long startTi, long endTi, Set<String> tags, long timeInterval, String durationLiteral, String meanFiled) {
        WhereQueryImpl<SelectQueryImpl> where = select().mean(meanFiled).as(meanFiled).from(database, measurement).where();
        buildQryWhereWithParam(where, param, tags);
        where.and(gte("time",ti(startTi, MILLISECONDS)))
                .and(lte("time",ti(endTi, MILLISECONDS)))
                .groupBy(time(timeInterval, durationLiteral));
        Query query = where;
        return convert(influxDB.query(query));
    }*/

    public List<Table<String, Object>> queryMeansBetweenTisWithGroupByTi(String measurement, Table<String, Object> param,
                            long startTi, long endTi, Set<String> tags, long timeInterval, String durationLiteral, Set<String> meanFileds) {
        SelectionQueryImpl select = select();
        for(String meanFiled : meanFileds) {
            select.mean(meanFiled).as(meanFiled);
        }
        WhereQueryImpl<SelectQueryImpl> where = select.from(database, measurement).where();
        buildQryWhereWithParam(where, param, tags);
        where.and(gte("time",ti(startTi, MILLISECONDS)))
                .and(lte("time",ti(endTi, MILLISECONDS)))
                .groupBy(time(timeInterval, durationLiteral));
        Query query = where;
        return convert(influxDB.query(query));
    }

    private void buildQryWhereWithParam(WhereQueryImpl<SelectQueryImpl> where, Table<String, Object>param, Set<String> tags) {
        for(ILink<String, Object> link : param) {
            if(!Text.isBlank(link.origin())) {
                if(!"time".equalsIgnoreCase(link.origin())) {
                    if(tags.contains(link.origin())) {
                        where.and(eq(link.origin(), ""+link.destination()));
                    } else {
                        where.and(eq(link.origin(), link.destination()));
                    }
                }
            }
        }
    }

    /**
     * 插入
     *
     * @param measurement
     *            表
     * @param tags
     *            标签
     * @param fields
     *            字段
     */
    public void insert(String measurement, Map<String, String> tags, Map<String, Object> fields, long time) {
        Point.Builder builder = Point.measurement(measurement);
        builder.tag(tags);
        builder.fields(fields);
        if (0 != time) {
            builder.time(time, TimeUnit.MILLISECONDS);
        } else {
            //builder.time(DateUtils.addHours(new Date(System.currentTimeMillis()), 8).getTime(), TimeUnit.MILLISECONDS);
            builder.time(System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }
        influxDB.write(database, retentionPolicy, builder.build());
    }

    public void insert(String measurement, Map<String, String> tags, Map<String, Object> fields) {
        insert(measurement, tags, fields, 0);
    }

    /**
     * 批量写入数据
     *
     * @param database
     *            数据库
     * @param consistency
     *            一致性
     * @param points
     *            要保存的数据（调用BatchPoints.lineProtocol()可得到一条record）
     */
    public void batchInsert(final String database, final InfluxDB.ConsistencyLevel consistency,
                            final Map<String, String> tags, final List<Point> points) {
        List<String> records = new ArrayList<String>();
        BatchPoints.Builder db = BatchPoints.database(database);
        for(String k : tags.keySet()) {
            db.tag(k, tags.get(k));
        }
        BatchPoints batchPoints = db.retentionPolicy(retentionPolicy).consistency(InfluxDB.ConsistencyLevel.ALL).build();
        points.stream().forEach(p-> {
            batchPoints.point(p);
        });
        records.add(batchPoints.lineProtocol());//这里只使用一个record,因为只构造一个tag
        influxDB.write(database, retentionPolicy, InfluxDB.ConsistencyLevel.ALL, records);
    }

    /**
     * 关闭数据库
     */
    public void close() {
        influxDB.close();
    }

    /**
     * 构建Point
     *
     * @param measurement
     * @param time
     * @param fields
     * @return
     */
    public Point pointBuilder(String measurement, long time, Map<String, String> tags, Map<String, Object> fields) {
        return Point.measurement(measurement).time(time, TimeUnit.MILLISECONDS).tag(tags).fields(fields).build();
    }

    private List<Table<String, Object>> convert(QueryResult results) {
        if (null == results || null == results.getResults() || 0 == results.getResults().size()) {
            return null;
        }
        List<Table<String, Object>> ret = new ArrayList<>();
        for (QueryResult.Result result : results.getResults()) {
            if(result.hasError()) {
                logger.error("query wrapper error:" + result.getError());
                continue;
            }
            List<QueryResult.Series> series = result.getSeries();
            if(null == series || 0 == series.size()) {
                return null;
            }
            for (QueryResult.Series serie : series) {
                List<List<Object>> values = serie.getValues();
                List<String> colums = serie.getColumns();
                for (List<Object> value : values) {
                    Table<String, Object> t = new Table<>();
                    for(int i=0; i<colums.size(); i++) {
                        t.put(colums.get(i), value.get(i));
                    }
                    ret.add(t);
                }
            }
        }
        return ret;
    }
}

